package skewreduce.pig;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.EOFException;
import java.net.URL;
import java.util.Map;

import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.io.BufferedPositionedInputStream;
import org.apache.pig.impl.logicalLayer.schema.Schema;

public class RecordStorage extends EaggLoadBase {
    private DataInputStream input;
    private long pos;
    private long end;

    public RecordStorage(String format) { super(format); }

    @Override
    public void bindTo(String fileName,
                       BufferedPositionedInputStream is,
                       long offset,
                       long end) throws IOException {
        this.pos = offset;
        this.end = end;

        int recSz = getMinRecordSize();
        if ( offset != 0 ) {
            long skip = recSz - ( offset % recSz );
            skip = ( skip == recSz ) ? 0 : skip;
            if ( skip > 0 ) is.skip(skip);
            pos += skip;
        }
        input = new DataInputStream(is);
    }

    @Override
    public Tuple getNext() throws IOException {
        if ( input == null || pos >= end ) {
            return null;
        }
        Tuple tuple = null;
        try {
            tuple = createTuple(input);
        } catch ( EOFException eof ) {
            System.err.printf("Partially written record? pos = %d, end = %d, record size = %d\n",pos,end,getMinRecordSize());
            tuple = null;
        }
        pos += getMinRecordSize();
        return tuple;
    }
}
